package com.winit.openapi.es.proxy;

import java.util.concurrent.Callable;
import java.util.concurrent.Future;

import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.Assert;

public class ElasticsearchClentFactoryBean implements FactoryBean<Client>, InitializingBean, DisposableBean {

    private static final Logger      logger = LoggerFactory.getLogger(ElasticsearchClentFactoryBean.class);

    protected Client                 client;

    protected Client                 proxyfiedClient;

    protected boolean                async  = false;

    protected ThreadPoolTaskExecutor taskExecutor;

    private String clusterName;

    private String nodes;

    public ElasticsearchClentFactoryBean(){
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        logger.info("Starting ElasticSearch client");

        if (async) {
            Assert.notNull(taskExecutor);
            Future<Client> future = taskExecutor.submit(new Callable<Client>() {

                @Override
                public Client call() throws Exception {
                    return initialize();
                }
            });

            ProxyFactory proxyFactory = new ProxyFactory();
            proxyFactory.setProxyTargetClass(true);
            proxyFactory.setTargetClass(Client.class);
            proxyFactory.addAdvice(new GenericInvocationHandler(future));
            proxyfiedClient = (Client) proxyFactory.getProxy();
        } else {
            client = initialize();
        }
    }

    private Client initialize() throws Exception {
        Settings settings = Settings.settingsBuilder().put("cluster.name", getClusterName()).build();

        String[] nodes = getNodes().split("\\,");
        Client client = TransportClient.builder()
            .settings(settings)
            .build()
            .addTransportAddresses(new TransportAddresses(nodes).getAddresses());
        return client;
    }

    @Override
    public Client getObject() throws Exception {
        return async ? proxyfiedClient : client;
    }

    @Override
    public Class<?> getObjectType() {
        return Client.class;
    }

    @Override
    public boolean isSingleton() {
        return true;
    }

    @Override
    public void destroy() throws Exception {
        try {
            logger.info("Closing ElasticSearch client");
            if (client != null) {
                client.close();
            }
        } catch (final Exception e) {
            logger.error("Error closing ElasticSearch client: ", e);
        }
    }

    public String getClusterName() {
        return clusterName;
    }

    public void setClusterName(String clusterName) {
        this.clusterName = clusterName;
    }

    public String getNodes() {
        return nodes;
    }

    public void setNodes(String nodes) {
        this.nodes = nodes;
    }

    public boolean isAsync() {
        return async;
    }

    public void setAsync(boolean async) {
        this.async = async;
    }

    public ThreadPoolTaskExecutor getTaskExecutor() {
        return taskExecutor;
    }

    public void setTaskExecutor(ThreadPoolTaskExecutor taskExecutor) {
        this.taskExecutor = taskExecutor;
    }

}
